package com.haoziqi.chapter_07;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.server.SessionTracker;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import javax.swing.*;

/**
 * description
 * created by A on 2021/3/11
 */
public class TimeWindowDemo {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.转化数据
        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("hadoop102", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        // 切分
                        String[] line = value.split(",");
                        return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));

                    }
                });
        //分组前开窗的话，会将所有组放在一个窗口
        //分组后开窗，每个Task的每个组是一个独立的窗口
        //keyBy两个参数的意思,第一个是传入的数据的类型,第二个指的是Key的类型
        KeyedStream<WaterSensor, String> keyByDS = sensorDS.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.getId();
            }
        });

        WindowedStream<WaterSensor, String, TimeWindow> window = keyByDS
               // .window(TumblingProcessingTimeWindows.of(Time.seconds(5))); //滚动窗口
        //timeWindow(Time.seconds(5)); //老版本写法
           //     .window(SlidingProcessingTimeWindows.of(Time.hours(1),Time.seconds(5)));//滑动窗口
            //滑动窗口适用场景，比如每5分钟，求最近一小时的数据
                //timeWindow(Time.hours(1),Time.seconds(5));
                //.window(ProcessingTimeSessionWindows.withGap(Time.seconds(3))); //会话窗口：动态间隔
                .window(ProcessingTimeSessionWindows.withDynamicGap(
                        new SessionWindowTimeGapExtractor<WaterSensor>() {
                            @Override
                            public long extract(WaterSensor element) {
                                return element.getTs()*1000L; //Flink里的时间戳单位都是毫秒
                            }
                        }
                ));
        //提交任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
